package com.bw.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TradeProvinceOrderBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.function.DimRichAsyncFunction;
import com.bw.gmall.realtime.common.util.ClickHouseUtil;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.TimeUnit;

public class DwsTradeProvinceOrderWindow extends BaseApp {

    public static void main(String[] args) throws Exception {
        new DwsTradeProvinceOrderWindow().start(10020,
                4,
                Constant.DWS_TRADE_PROVINCE_ORDER_WINDOW,
                Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 1. 数据清洗
        SingleOutputStreamOperator<JSONObject> etlStream = stream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {

                try {
                    JSONObject jsonObject = JSON.parseObject(s);
                    Long ts = jsonObject.getLong("ts");
                    if (ts != null) {
                        jsonObject.put("ts",ts*1000);
                        collector.collect(jsonObject);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        // 2. 加水位線 分組 數據校驗
        SingleOutputStreamOperator<TradeProvinceOrderBean> processStream = etlStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject jsonObject, long l) {
                return jsonObject.getLong("ts");
            }
        })).keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getString("id");
            }
            /**
             *  1   100      100
             *  1   120--20  120
             */
        }).process(new KeyedProcessFunction<String, JSONObject, TradeProvinceOrderBean>() {
            ValueState<BigDecimal> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<BigDecimal> stateDescriptor = new ValueStateDescriptor<>("splitTotalAmount", BigDecimal.class);
                // 状态过期时间
                stateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(30L)).build());
                valueState = getRuntimeContext().getState(stateDescriptor);
            }

            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, TradeProvinceOrderBean>.Context context, Collector<TradeProvinceOrderBean> collector) throws Exception {
                // 订单金额
                BigDecimal splitTotalAmount = jsonObject.getBigDecimal("split_total_amount");

                //状态里面的金额
                BigDecimal value = valueState.value();
                // 初始化为0
                value = value == null ? new BigDecimal(0) : value;

                // 用来装订单号
                Set<String> orderIdSets = new HashSet<>();
                orderIdSets.add(jsonObject.getString("order_id"));

                collector.collect(TradeProvinceOrderBean.builder()
                        .provinceId(jsonObject.getString("province_id"))
                        .orderAmount(splitTotalAmount.subtract(value))
                        .orderDetailId(jsonObject.getString("id"))
                        .orderIdSet(orderIdSets)
                        .build());

                valueState.update(splitTotalAmount);
            }
        });
        // 3、以省份分组，开窗  聚合
        SingleOutputStreamOperator<TradeProvinceOrderBean> reduceStream = processStream.keyBy(new KeySelector<TradeProvinceOrderBean, String>() {
            @Override
            public String getKey(TradeProvinceOrderBean tradeProvinceOrderBean) throws Exception {
                return tradeProvinceOrderBean.getProvinceId();
            }
        }).window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10))).reduce(new ReduceFunction<TradeProvinceOrderBean>() {
            @Override
            public TradeProvinceOrderBean reduce(TradeProvinceOrderBean t1, TradeProvinceOrderBean t2) throws Exception {
                // 求金额
                t1.setOrderAmount(t1.getOrderAmount().add(t2.getOrderAmount()));
                // 求订单数
                t1.getOrderIdSet().addAll(t2.getOrderIdSet());

                return t1;
            }
        }, new WindowFunction<TradeProvinceOrderBean, TradeProvinceOrderBean, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow timeWindow, Iterable<TradeProvinceOrderBean> iterable, Collector<TradeProvinceOrderBean> collector) throws Exception {
                String start = DateFormatUtil.tsToDateTime(timeWindow.getStart());
                String end = DateFormatUtil.tsToDateTime(timeWindow.getEnd());
                Iterator<TradeProvinceOrderBean> iterator = iterable.iterator();
                while (iterator.hasNext()) {
                    TradeProvinceOrderBean tradeProvinceOrderBean = iterator.next();
                    tradeProvinceOrderBean.setOrderCount((long) tradeProvinceOrderBean.getOrderIdSet().size());
                    tradeProvinceOrderBean.setStt(start);
                    tradeProvinceOrderBean.setEdt(end);
                    tradeProvinceOrderBean.setTs(System.currentTimeMillis());
                    tradeProvinceOrderBean.setCurDate(DateFormatUtil.tsToDate(System.currentTimeMillis()));
                    collector.collect(tradeProvinceOrderBean);
                }
            }
        });

        // 5、关联维度消息
        SingleOutputStreamOperator<TradeProvinceOrderBean> dimStream = AsyncDataStream.unorderedWait(reduceStream, new DimRichAsyncFunction<TradeProvinceOrderBean>() {
            @Override
            public String getDimTableName() {
                return "dim_base_province";
            }

            @Override
            public String getId(TradeProvinceOrderBean tradeProvinceOrderBean) {
                return tradeProvinceOrderBean.getProvinceId();
            }

            @Override
            public void setDimInfo(TradeProvinceOrderBean tradeProvinceOrderBean, JSONObject dimInfo) {
                String provinceName = dimInfo.getString("name");
                tradeProvinceOrderBean.setProvinceName(provinceName);

            }
        }, 60, TimeUnit.SECONDS);
        //4、写到CK
        dimStream.print();
        /**
         * create table province_2203A(
         *   stt DateTime,
         *   edt DateTime,
         *   cur_date String,
         *   province_id String,
         *   province_name String,
         *   order_count UInt64,
         *   order_amount Float64,
         *   ts UInt64,
         *   order_detail_id String
         * )engine=ReplacingMergeTree(ts)  partition by toYYYYMMDD(stt)
         * order by ( stt,edt,province_id,province_name)
         */
        dimStream.addSink(ClickHouseUtil.sink("insert into province_2203A (stt,\n" +
                "     edt,\n" +
                "     cur_date,\n" +
                "     province_id,\n" +
                "     province_name,\n" +
                "     order_count,\n" +
                "     order_amount,\n" +
                "     ts,\n" +
                "     order_detail_id) values (?,?,?,?,?,?,?,?,?)"));

    }
}
